package com.pw.study.flink.exercise;


import com.pw.study.flink.entities.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author: linux_future
 * @description： 布隆过滤器存在假阳性的概率，不适用于任何要求100%准确率的情境；
 * 由于它不存在假阴性问题，所以用作“不存在”逻辑的处理时有奇效，比如可以用来作为缓存系统（如Redis）的缓冲，防止缓存穿透
 * @since: 2022/3/14
 **/
public class ExeBloomFilter {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //读文件
        SingleOutputStreamOperator<UserBehavior> data = env
                .readTextFile("data/file/UserBehavior.csv")
                .map(line -> { // 对数据切割, 然后封装到POJO中
                    String[] split = line.split(",");
                    return new UserBehavior(Long.valueOf(split[0]), Long.valueOf(split[1]), Integer.valueOf(split[2]), split[3], Long.valueOf(split[4]));
                });

        // 创建WatermarkStrategy
        WatermarkStrategy<UserBehavior> wms = WatermarkStrategy
                .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                    @Override
                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                        return element.getTimestamp() * 1000L;
                    }
                });
        //数据处理
        data.filter(pv -> pv.getBehavior().equals("pv")).assignTimestampsAndWatermarks(wms)
                .keyBy(UserBehavior::getBehavior)//全局过滤（相当于没做过滤，但是必须keyby）
                .window(TumblingEventTimeWindows.of(Time.seconds(3)))
                .process(new ProcessWindowFunction<UserBehavior, String, String, TimeWindow>() {
                    private ValueState<Long> countState;
                    //布隆过滤器
                    private ValueState<BloomFilter<Long>> bfState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        bfState = getRuntimeContext().getState(new ValueStateDescriptor<BloomFilter<Long>>("bfState", TypeInformation.of(new TypeHint<BloomFilter<Long>>() {
                        })));
                        countState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("countState", Long.class));
                    }

                    @Override
                    public void process(String key, Context ctx, Iterable<UserBehavior> elements, Collector<String> out) throws Exception {
                        ValueState<BloomFilter<Long>> bfState = this.bfState;
                        if (bfState.value() == null) {
                            // 在状态中初始化一个布隆过滤器
                            // 参数1: 漏斗, 存储的类型
                            // 参数2: 期望插入的元素总个数
                            // 参数3: 期望的误判率(假阳性率)
                            BloomFilter<Long> bf = BloomFilter.create(Funnels.longFunnel(), 100*100000, 0.001);
                            bfState.update(bf);
                           countState.update(0L);
                        }
                        BloomFilter<Long> bf = bfState.value();
                        for (UserBehavior uv : elements) {
                            if (!bf.mightContain(uv.getUserId())) {
                                // 不存在 计数+1
                                countState.update(countState.value() + 1L);
                                // 记录这个用户di, 表示来过
                                bf.put(uv.getUserId());
                            }

                        }
                            out.collect("窗口: " + ctx.window() + " 的uv是: " + countState.value());
                    }
                }).print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
